package rocketmq

import (
	"errors"
	"github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
	"shen/gotask/rocketmq/config"
	"shen/gotask/rocketmq/logs"
	"time"
)

var (
	RocketmqProducer map[string]golang.Producer
)

func GetProducer(topic string) (error, golang.Producer) {
	if RocketmqProducer == nil {
		RocketmqProducer = make(map[string]golang.Producer)
	}
	var producer golang.Producer
	producer, ok := RocketmqProducer[topic]
	if !ok {
		pro, err := golang.NewProducer(&golang.Config{
			Endpoint: config.Config.Rocketmq.Endpoint,
			Credentials: &credentials.SessionCredentials{
				AccessKey:    config.Config.Rocketmq.AccessKey,
				AccessSecret: config.Config.Rocketmq.AccessSecret,
			},
		},
			golang.WithTopics(topic),
		)
		if err != nil {
			return err, nil
		}
		producer = pro
		err2 := producer.Start()
		if err2 != nil {
			return err2, nil
		}
		RocketmqProducer[topic] = producer
	} else {
		if _, ok := producer.(golang.Producer); !ok {
			delete(RocketmqProducer, topic)
			return errors.New("producer is not golang.Producer"), nil
		}
	}

	return nil, producer
}

func StopProducer(topic string) error {
	defer func() {
		if err := recover(); err != nil {
			logs.NewLogs().Error("关闭rocketmq客户端失败")
		}
	}()
	if len(RocketmqProducer) > 0 {
		producer, ok := RocketmqProducer[topic]
		if ok {
			if _, isProducer := producer.(golang.Producer); isProducer {
				producer.GracefulStop()
			}
			delete(RocketmqProducer, topic)
		}
	}
	return nil
}

func CreateConsumer(topic, groupName, expression string, expressionType golang.FilterExpressionType) (golang.SimpleConsumer, error) {
	awaitDuration := time.Second * 5
	simpleConsumer, err := golang.NewSimpleConsumer(&golang.Config{
		Endpoint:      config.Config.Rocketmq.Endpoint,
		ConsumerGroup: groupName,
		Credentials: &credentials.SessionCredentials{
			AccessKey:    config.Config.Rocketmq.AccessKey,
			AccessSecret: config.Config.Rocketmq.AccessSecret,
		},
	},
		golang.WithAwaitDuration(awaitDuration),
		golang.WithSubscriptionExpressions(map[string]*golang.FilterExpression{
			topic: golang.NewFilterExpressionWithType(expression, expressionType),
		}),
	)
	if err != nil {
		return nil, errors.New("NewSimpleConsumer err:%v" + err.Error())
	}
	err = simpleConsumer.Start()
	if err != nil {
		return nil, errors.New("MustStartConsumer err:" + err.Error())
	}
	return simpleConsumer, nil
}
